"""
AkShare数据客户端

获取金融数据
"""

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Any, Tuple
import logging
import time
import threading
import backoff
from ..models.market_data import (
    StockPrice, FinancialMetrics, MarketData, NewsItem,
    TechnicalIndicators, DataCollectionSummary, StockBasicInfo
)
from ..utils.data_cache import cached, CacheLevel, get_cache_manager, invalidate_cache

logger = logging.getLogger(__name__)


class AkShareClient:
    """AkShare数据客户端"""

    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        """单例模式实现"""
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super(AkShareClient, cls).__new__(cls)
                    cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        """初始化AkShare客户端（单例模式）"""
        if self._initialized:
            return

        self.session_start = datetime.now()
        self.cache_manager = get_cache_manager()
        self._initialized = True
        logger.info("AkShare客户端初始化完成（单例模式）")
        logger.info("数据缓存管理器已启用")
    
    @cached(cache_level=CacheLevel.MEDIUM)  # 30分钟缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_stock_hist_data(
        self,
        symbol: str,
        start_date: str = "20200101",
        end_date: Optional[str] = None,
        adjust: str = ""
    ) -> List[StockPrice]:
        """
        获取股票历史数据

        Args:
            symbol: 股票代码，如 "000001"
            start_date: 开始日期，格式 "YYYYMMDD"
            end_date: 结束日期，格式 "YYYYMMDD"，默认为今天
            adjust: 复权类型，"" 不复权，"qfq" 前复权，"hfq" 后复权

        Returns:
            List[StockPrice]: 股票价格数据列表
        """
        try:
            if end_date is None:
                end_date = datetime.now().strftime("%Y%m%d")

            logger.info(f"获取股票 {symbol} 历史数据...")
            logger.info(f"开始日期: {start_date}, 结束日期: {end_date}")

            def get_and_process_data(start_dt: str, end_dt: str) -> List[StockPrice]:
                """获取并处理数据的内部函数"""
                df = ak.stock_zh_a_hist(
                    symbol=symbol,
                    period="daily",
                    start_date=start_dt,
                    end_date=end_dt,
                    adjust=adjust
                )

                if df.empty:
                    return []

                # 转换为StockPrice模型
                stock_prices = []
                for _, row in df.iterrows():
                    try:
                        stock_price = StockPrice(
                            date=pd.to_datetime(row['日期']),
                            symbol=str(row['股票代码']),
                            open=float(row['开盘']),
                            high=float(row['最高']),
                            low=float(row['最低']),
                            close=float(row['收盘']),
                            volume=int(row['成交量']),
                            amount=float(row.get('成交额', 0)) if pd.notna(row.get('成交额')) else None,
                            pct_change=float(row.get('涨跌幅', 0)) if pd.notna(row.get('涨跌幅')) else None,
                            change_amount=float(row.get('涨跌额', 0)) if pd.notna(row.get('涨跌额')) else None,
                            amplitude=float(row.get('振幅', 0)) if pd.notna(row.get('振幅')) else None,
                            turnover=float(row.get('换手率', 0)) if pd.notna(row.get('换手率')) else None
                        )
                        stock_prices.append(stock_price)
                    except Exception as e:
                        logger.warning(f"解析股票数据行时出错: {e}")
                        continue

                return stock_prices

            # 获取历史数据
            stock_prices = get_and_process_data(start_date, end_date)

            if not stock_prices:
                logger.warning(f"未获取到股票 {symbol} 的历史数据")
                return []

            # === 智能数据检查 ===
            min_required_days = 120  # 至少需要120个交易日的数据
            if len(stock_prices) < min_required_days:
                logger.warning(f"数据量不足（{len(stock_prices)} 天），技术指标计算可能不准确")
                logger.info("尝试扩大时间范围获取更多数据...")

                # 扩大时间范围到2年
                try:
                    # 计算2年前的日期
                    end_dt = datetime.strptime(end_date, "%Y%m%d")
                    extended_start_dt = end_dt - timedelta(days=730)
                    extended_start_date = extended_start_dt.strftime("%Y%m%d")

                    logger.info(f"扩大时间范围: {extended_start_date} 到 {end_date}")
                    extended_stock_prices = get_and_process_data(extended_start_date, end_date)

                    if len(extended_stock_prices) >= len(stock_prices):
                        stock_prices = extended_stock_prices
                        logger.info(f"✓ 扩大时间范围后获取到 {len(stock_prices)} 条数据")

                    if len(stock_prices) < min_required_days:
                        logger.warning(f"即使扩大时间范围，数据量仍然不足（{len(stock_prices)} 天）")
                        logger.warning("部分高级技术指标可能无法准确计算")

                except Exception as e:
                    logger.warning(f"扩大时间范围时出错: {e}")

            # 按日期排序
            stock_prices.sort(key=lambda x: x.date)

            logger.info(f"✓ 成功获取股票 {symbol} 历史数据 {len(stock_prices)} 条")
            return stock_prices

        except Exception as e:
            logger.error(f"获取股票历史数据失败: {e}")
            raise
    
    @cached(cache_level=CacheLevel.MEDIUM)  # 30分钟缓存-后续不缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_stock_info(self, symbol: str) -> Optional[StockBasicInfo]:
        """
        获取股票基本信息

        Args:
            symbol: 股票代码

        Returns:
            Optional[Dict]: 股票基本信息字典
        """
        try:
            # 使用akshare的stock_individual_info_em接口获取股票基本信息
            info_data = ak.stock_individual_info_em(symbol=symbol)

            if info_data is None or info_data.empty:
                logger.warning(f"未获取到股票 {symbol} 的基本信息")
                return None

            info_dict = dict(zip(info_data['item'], info_data['value']))

            # 构建返回的股票基本信息
            stock_info = {
                'symbol': symbol,
                'name': info_dict.get('股票简称'),
                'latest_price': self._safe_float(info_dict.get('最新')),
                'total_shares': self._safe_float(info_dict.get('总股本')),
                'float_shares': self._safe_float(info_dict.get('流通股')),
                'total_market_cap': self._safe_float(info_dict.get('总市值')),
                'float_market_cap': self._safe_float(info_dict.get('流通市值')),
                'industry': info_dict.get('行业'),
                'listing_date': info_dict.get('上市时间'),
                'update_time': datetime.now()
            }

            logger.info(f"成功获取股票 {symbol} 的基本信息: {stock_info['name']}")
            return stock_info

        except Exception as e:
            logger.error(f"获取股票 {symbol} 基本信息失败: {e}")
            return None

    @cached(cache_level=CacheLevel.SHORT)  # 5分钟缓存
    def get_stock_realtime_data(self, symbol: str) -> Optional[MarketData]:
        """
        获取股票实时数据

        Args:
            symbol: 股票代码

        Returns:
            Optional[MarketData]: 实时市场数据
        """
        # 接口优先级列表
        interfaces = [
            # ("历史数据最新", self._get_realtime_from_hist),
            # ("新浪财经", self._get_realtime_from_sina),
            ("东方财富", self._get_realtime_from_em)
        ]

        for interface_name, interface_func in interfaces:
            try:
                logger.info(f"尝试使用 {interface_name} 接口获取股票 {symbol} 实时数据...")
                result = interface_func(symbol)
                if result:
                    logger.info(f"✅ 成功通过 {interface_name} 接口获取股票 {symbol} 实时数据")
                    return result
                else:
                    logger.warning(f"❌ {interface_name} 接口返回空数据")
            except Exception as e:
                logger.warning(f"❌ {interface_name} 接口失败: {e}")
                continue

        logger.error(f"所有接口都失败，无法获取股票 {symbol} 的实时数据")
        return None

    # def _get_realtime_from_hist(self, symbol: str) -> Optional[MarketData]:
    #     """通过历史数据获取数据"""
    #     try:
    #         # 获取最近2天的数据，取最新的一条
    #         end_date = datetime.now().strftime("%Y%m%d")
    #         start_date = (datetime.now() - timedelta(days=5)).strftime("%Y%m%d")

    #         df = ak.stock_zh_a_hist(
    #             symbol=symbol,
    #             period="daily",
    #             start_date=start_date,
    #             end_date=end_date,
    #             adjust=""
    #         )

    #         if df.empty:
    #             return None

    #         # 取最新的一条数据
    #         latest = df.iloc[-1]

    #         return MarketData(
    #             ticker=symbol,
    #             name=f"股票{symbol}",  # 历史数据接口通常不包含名称
    #             current_price=float(latest.get('收盘', 0)) if pd.notna(latest.get('收盘')) else None,
    #             market_cap=None,  # 历史数据接口不包含市值
    #             volume=int(latest.get('成交量', 0)) if pd.notna(latest.get('成交量')) else None,
    #             fifty_two_week_high=float(latest.get('最高', 0)) if pd.notna(latest.get('最高')) else None,
    #             fifty_two_week_low=float(latest.get('最低', 0)) if pd.notna(latest.get('最低')) else None
    #         )

    #     except Exception as e:
    #         logger.warning(f"历史数据接口获取失败: {e}")
    #         return None

    # def _get_realtime_from_sina(self, symbol: str) -> Optional[MarketData]:
    #     """通过新浪财经接口获取数据）"""
    #     try:
    #         # 新浪接口主要提供指数数据，对个股支持有限
    #         # 这里作为备用方案，实际可能需要调整
    #         df = ak.stock_zh_index_spot_sina()

    #         # 尝试查找相关数据（新浪接口主要是指数）
    #         if df.empty:
    #             return None

    #         # 由于新浪接口主要是指数，这里返回None让系统尝试下一个接口
    #         return None

    #     except Exception as e:
    #         logger.warning(f"新浪财经接口获取失败: {e}")
    #         return None

    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def _get_realtime_from_em(self, symbol: str) -> Optional[MarketData]:
        """通过东方财富接口获取数据（使用所有返回字段）获取全部股票数据再获取指定股票数据"""
        try:
            # 设置较短的超时时间，避免长时间等待
            # df = ak.stock_zh_a_spot_em()
            df = self._get_cached_realtime_data()
            
            if df is None or df.empty:
                logger.warning("获取实时行情数据失败")
                return None

            # 查找指定股票
            stock_data = df[df['代码'] == symbol]
            if stock_data.empty:
                logger.warning(f"未找到股票 {symbol} 的实时数据")
                return None

            row = stock_data.iloc[0]

            # 构建完整的MarketData，使用所有可用字段
            market_data = MarketData(
                # 基础信息
                symbol=symbol,
                
                # 价格数据
                current_price=self._safe_float(row.get('最新价')),
                open_price=self._safe_float(row.get('今开')),
                high_price=self._safe_float(row.get('最高')),
                low_price=self._safe_float(row.get('最低')),
                prev_close=self._safe_float(row.get('昨收')),
                
                # 涨跌数据
                pct_change=self._safe_float(row.get('涨跌幅')),
                change_amount=self._safe_float(row.get('涨跌额')),
                amplitude=self._safe_float(row.get('振幅')),
                speed=self._safe_float(row.get('涨速')),
                change_5min=self._safe_float(row.get('5分钟涨跌')),
                change_60d=self._safe_float(row.get('60日涨跌幅')),
                change_ytd=self._safe_float(row.get('年初至今涨跌幅')),
                
                # 成交数据
                volume=int(row.get('成交量', 0)) if pd.notna(row.get('成交量')) else None,
                turnover=self._safe_float(row.get('成交额')),
                turnover_rate=self._safe_float(row.get('换手率')),
                volume_ratio=self._safe_float(row.get('量比')),
                
                # 估值数据
                pe_ratio=self._safe_float(row.get('市盈率-动态')),
                pb_ratio=self._safe_float(row.get('市净率')),
                market_cap=self._safe_float(row.get('总市值')),
                float_market_cap=self._safe_float(row.get('流通市值')),
                
                # 兼容性字段（暂时设为None，因为该接口不提供52周数据）
                fifty_two_week_high=None,
                fifty_two_week_low=None,
                average_volume=None,
                beta=None,
                dividend_yield=None
            )

            logger.info(f"✓ 成功获取股票 {symbol} 完整实时数据（包含所有字段）")
            return market_data

        except Exception as e:
            logger.warning(f"东方财富接口获取失败: {e}")
            return None

    @cached(cache_level=CacheLevel.MEDIUM)  # 30分钟缓存
    def _get_cached_realtime_data(self) -> Optional[pd.DataFrame]:
        """获取缓存的完整实时行情数据"""
        try:
            import os
            
            # 确保data目录存在
            data_dir = "data"
            cache_file = os.path.join(data_dir, "stock_zh_a_spot_em.csv")
            
            if not os.path.exists(data_dir):
                os.makedirs(data_dir)
                logger.info(f"创建数据目录: {data_dir}")
            
            df = None
            need_refresh = True
            
            # 检查缓存文件是否存在且有效
            if os.path.exists(cache_file):
                try:
                    # 指定代码列为字符串类型，防止000001变成1
                    df = pd.read_csv(cache_file, dtype={'代码': str})
                    
                    # 检查数据是否有效（非空且有时间戳列）
                    if not df.empty and '时间戳' in df.columns:
                        # 获取最新的时间戳
                        latest_timestamp_str = df.iloc[-1]['时间戳']
                        latest_timestamp = pd.to_datetime(latest_timestamp_str)
                        
                        # 判断数据是否在1小时内（缓存有效期）
                        cache_age = datetime.now() - latest_timestamp
                        if cache_age < timedelta(hours=1):
                            need_refresh = False
                            logger.info(f"使用缓存数据，缓存年龄: {cache_age}")
                        else:
                            logger.info(f"缓存数据过期，缓存年龄: {cache_age}，需要刷新")
                    else:
                        logger.warning("缓存文件数据格式无效，需要刷新")
                        
                except Exception as e:
                    logger.warning(f"读取缓存文件失败: {e}，将重新获取数据")
                    need_refresh = True
            else:
                logger.info("缓存文件不存在，将获取新数据")
            
            # 如果需要刷新数据
            if need_refresh:
                logger.info("正在获取最新的全市场实时行情数据...")
                df = ak.stock_zh_a_spot_em()
                
                if df is not None and not df.empty:
                    # 确保代码列是字符串格式（重要！）
                    if '代码' in df.columns:
                        df['代码'] = df['代码'].astype(str).str.zfill(6)  # 补齐6位，如1->000001
                    
                    # 添加时间戳列
                    df.insert(0, "时间戳", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
                    
                    # 保存到缓存文件
                    df.to_csv(cache_file, index=False)
                    logger.info(f"✓ 成功获取并缓存实时数据（{len(df)} 条记录）")
                else:
                    logger.error("获取实时数据失败，返回空DataFrame")
                    return None
            
            # 确保代码列是正确的字符串格式（双重保险）
            if df is not None and '代码' in df.columns:
                df['代码'] = df['代码'].astype(str).str.zfill(6)
            
            # 移除时间戳列（返回给调用者时不需要）
            if df is not None and '时间戳' in df.columns:
                df = df.drop('时间戳', axis=1)
            
            return df
            
        except Exception as e:
            logger.warning(f"获取完整实时数据失败: {e}")
            return None
    
    @cached(cache_level=CacheLevel.MEDIUM)  # 30分钟缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_financial_indicators(self, symbol: str) -> Optional[FinancialMetrics]:
        """
        获取财务指标数据

        Args:
            symbol: 股票代码

        Returns:
            Optional[FinancialMetrics]: 财务指标数据
        """
        try:
            logger.info(f"获取股票 {symbol} 的财务指标...")

            # 1. 获取完整的实时行情数据（只调用一次）
            logger.info("获取实时行情数据...")
            realtime_data = self._get_cached_realtime_data()
            if realtime_data is None or realtime_data.empty:
                logger.warning("无法获取完整实时行情数据")
                return None

            # 查找指定股票的数据
            stock_data = realtime_data[realtime_data['代码'] == symbol]
            if stock_data.empty:
                logger.warning(f"未找到股票 {symbol} 的实时行情")
                return None

            stock_data = stock_data.iloc[0]
            logger.info("✓ 实时行情数据获取成功")

            # 2. 获取财务指标数据
            logger.info("获取财务指标数据...")
            current_year = datetime.now().year
            financial_data = ak.stock_financial_analysis_indicator(
                symbol=symbol, start_year=str(current_year-1))
            if financial_data is None or financial_data.empty:
                logger.warning("无法获取财务指标数据")
                return None

            latest_financial = financial_data.iloc[-1]
            logger.info("✓ 财务指标数据获取成功")

            # 3. 获取利润表数据（用于计算市销率）
            logger.info("获取利润表数据...")
            try:
                # 使用自动前缀转换函数
                stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
                income_statement = ak.stock_financial_report_sina(
                    stock=stock_symbol_with_prefix, symbol="利润表")
                if not income_statement.empty:
                    latest_income = income_statement.iloc[0]
                    logger.info("✓ 利润表数据获取成功")
                else:
                    logger.warning("利润表数据为空")
                    latest_income = pd.Series()
            except Exception as e:
                logger.warning(f"获取利润表数据失败: {e}")
                latest_income = pd.Series()

            # 4. 构建完整的财务指标
            def convert_percentage(value: float) -> float:
                """将百分比值转换为小数"""
                try:
                    return float(value) / 100.0 if value is not None else 0.0
                except:
                    return 0.0

            # 5. 获取完整的财务报表数据（新增）
            logger.info("获取完整财务报表数据...")
            try:
                financial_statements = self.get_financial_statements(symbol)
                if financial_statements and len(financial_statements) >= 2:
                    current_statement = financial_statements[0]
                    previous_statement = financial_statements[1]
                    logger.info("✓ 财务报表数据获取成功")
                else:
                    logger.warning("财务报表数据不完整，使用默认值")
                    current_statement = {}
                    previous_statement = {}
            except Exception as e:
                logger.warning(f"获取财务报表数据失败: {e}")
                current_statement = {}
                previous_statement = {}

            financial_metrics = FinancialMetrics(
                # 市场数据
                market_cap=self._safe_float(stock_data.get("总市值", 0)),
                float_market_cap=self._safe_float(stock_data.get("流通市值", 0)),

                # 盈利数据
                revenue=self._safe_float(latest_income.get("营业总收入", 0)),
                net_income=self._safe_float(latest_income.get("净利润", 0)),
                return_on_equity=convert_percentage(latest_financial.get("净资产收益率(%)", 0)),
                net_margin=convert_percentage(latest_financial.get("销售净利率(%)", 0)),
                operating_margin=convert_percentage(latest_financial.get("营业利润率(%)", 0)),

                # 增长指标
                revenue_growth=convert_percentage(latest_financial.get("主营业务收入增长率(%)", 0)),
                earnings_growth=convert_percentage(latest_financial.get("净利润增长率(%)", 0)),
                book_value_growth=convert_percentage(latest_financial.get("净资产增长率(%)", 0)),

                # 财务健康指标
                current_ratio=self._safe_float(latest_financial.get("流动比率", 0)),
                debt_to_equity=convert_percentage(latest_financial.get("资产负债率(%)", 0)),
                free_cash_flow_per_share=self._safe_float(latest_financial.get("每股经营性现金流(元)", 0)),
                earnings_per_share=self._safe_float(latest_financial.get("加权每股收益(元)", 0)),

                # 估值比率
                pe_ratio=self._safe_float(stock_data.get("市盈率-动态", 0)),
                price_to_book=self._safe_float(stock_data.get("市净率", 0)),
                price_to_sales=self._safe_float(stock_data.get("总市值", 0)) / self._safe_float(latest_income.get("营业总收入", 1)) if self._safe_float(latest_income.get("营业总收入", 0)) and self._safe_float(latest_income.get("营业总收入", 0)) > 0 else 0,

                # === 财务报表数据（当期） ===
                operating_revenue=self._safe_float(current_statement.get("operating_revenue", 0)),
                operating_profit=self._safe_float(current_statement.get("operating_profit", 0)),
                working_capital=self._safe_float(current_statement.get("working_capital", 0)),
                depreciation_and_amortization=self._safe_float(current_statement.get("depreciation_and_amortization", 0)),
                capital_expenditure=self._safe_float(current_statement.get("capital_expenditure", 0)),
                free_cash_flow=self._safe_float(current_statement.get("free_cash_flow", 0)),

                # === 财务报表数据（上期） ===
                previous_operating_revenue=self._safe_float(previous_statement.get("operating_revenue", 0)),
                previous_net_income=self._safe_float(previous_statement.get("net_income", 0)),
                previous_operating_profit=self._safe_float(previous_statement.get("operating_profit", 0)),
                previous_working_capital=self._safe_float(previous_statement.get("working_capital", 0)),
                previous_depreciation_and_amortization=self._safe_float(previous_statement.get("depreciation_and_amortization", 0)),
                previous_capital_expenditure=self._safe_float(previous_statement.get("capital_expenditure", 0)),
                previous_free_cash_flow=self._safe_float(previous_statement.get("free_cash_flow", 0))
            )

            logger.info(f"成功获取股票 {symbol} 完整财务指标（含财务报表数据）")
            return financial_metrics

        except Exception as e:
            logger.error(f"获取财务指标失败: {e}")
            return None
    
    @cached(cache_level=CacheLevel.MEDIUM)  # 30分钟缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_stock_news(self, symbol: str, count: int = 10) -> List[NewsItem]:
        """
        获取股票相关新闻
        
        Args:
            symbol: 股票代码
            count: 新闻数量
            
        Returns:
            List[NewsItem]: 新闻列表
        """
        try:
            df = ak.stock_news_em(symbol=symbol)
            
            if df.empty:
                logger.warning(f"未获取到股票 {symbol} 的新闻数据")
                return []
            
            news_items = []
            for _, row in df.head(count).iterrows():
                try:
                    news_item = NewsItem(
                        title=row.get('新闻标题', ''),
                        content=row.get('新闻内容', ''),
                        publish_time=pd.to_datetime(row.get('发布时间', datetime.now())),
                        source=row.get('新闻来源', ''),
                        url=row.get('新闻链接', '')
                    )
                    news_items.append(news_item)
                except Exception as e:
                    logger.warning(f"解析新闻数据时出错: {e}")
                    continue
            
            logger.info(f"成功获取股票 {symbol} 新闻 {len(news_items)} 条")
            return news_items
            
        except Exception as e:
            logger.error(f"获取股票新闻失败: {e}")
            return []
    
    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_macro_data(self, indicator: str = "cpi") -> List[Dict[str, Any]]:
        """
        获取宏观经济数据
        
        Args:
            indicator: 指标类型，如 "cpi", "pmi", "gdp"
            
        Returns:
            List[Dict[str, Any]]: 宏观数据列表
        """
        try:
            # 根据指标类型选择对应的akshare函数
            if indicator.lower() == "cpi":
                df = ak.macro_china_cpi_monthly()
            elif indicator.lower() == "pmi":
                df = ak.macro_china_pmi()
            elif indicator.lower() == "gdp":
                df = ak.macro_china_gdp()
            else:
                logger.warning(f"不支持的宏观指标: {indicator}")
                return []

            if df.empty:
                return []

            # 转换为字典列表
            macro_data = df.to_dict('records')
            logger.info(f"成功获取宏观数据 {indicator}: {len(macro_data)} 条")
            return macro_data

        except Exception as e:
            logger.error(f"获取宏观数据失败: {e}")
            return []

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_macro_indicators(self, analysis_scope: str = "china") -> Dict[str, Any]:
        """
        获取宏观经济指标

        Args:
            analysis_scope: 分析范围 (china/global)

        Returns:
            Dict[str, Any]: 宏观指标数据
        """
        try:
            logger.info(f"获取{analysis_scope}宏观经济指标")

            macro_indicators = {}

            if analysis_scope.lower() == "china":
                # 获取中国主要宏观指标
                try:
                    # CPI数据
                    cpi_data = ak.macro_china_cpi_monthly()
                    if not cpi_data.empty:
                        latest_cpi = cpi_data.iloc[0]
                        macro_indicators["cpi"] = {
                            "value": latest_cpi.get("今值", 0),
                            "date": latest_cpi.get("日期", ""),
                            "previous": latest_cpi.get("前值", 0)
                        }
                except Exception as e:
                    logger.warning(f"获取CPI数据失败: {e}")

                try:
                    # PMI数据
                    pmi_data = ak.macro_china_pmi_yearly()
                    if not pmi_data.empty:
                        latest_pmi = pmi_data.iloc[0]
                        macro_indicators["pmi"] = {
                            "value": latest_pmi.get("今值", 0),
                            "date": latest_pmi.get("日期", ""),
                            "previous": latest_pmi.get("前值", 0)
                        }
                except Exception as e:
                    logger.warning(f"获取PMI数据失败: {e}")

                try:
                    # GDP数据
                    gdp_data = ak.macro_china_gdp_yearly()
                    if not gdp_data.empty:
                        latest_gdp = gdp_data.iloc[0]
                        macro_indicators["gdp"] = {
                            "value": latest_gdp.get("今值", 0),
                            "date": latest_gdp.get("日期", ""),
                            "previous": latest_gdp.get("前值", 0)
                        }
                except Exception as e:
                    logger.warning(f"获取GDP数据失败: {e}")

                try:
                    # M2货币供应量
                    m2_data = ak.macro_china_m2_yearly()
                    if not m2_data.empty:
                        latest_m2 = m2_data.iloc[0]
                        macro_indicators["m2"] = {
                            "value": latest_m2.get("今值", 0),
                            "date": latest_m2.get("日期", ""),
                            "previous": latest_m2.get("前值", 0)
                        }
                except Exception as e:
                    logger.warning(f"获取M2数据失败: {e}")

                try:
                    # 失业率
                    unemployment_data = ak.macro_china_urban_unemployment()
                    if not unemployment_data.empty:
                        latest_unemployment = unemployment_data.iloc[0]
                        macro_indicators["unemployment"] = {
                            "value": latest_unemployment.get("value", 0),
                            "date": latest_unemployment.get("date", ""),
                            "item": latest_unemployment.get("item", "")
                        }
                except Exception as e:
                    logger.warning(f"获取失业率数据失败: {e}")

            logger.info(f"成功获取{analysis_scope}宏观指标: {len(macro_indicators)}项")
            return macro_indicators

        except Exception as e:
            logger.error(f"获取宏观指标失败: {e}")
            return {}

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_monetary_policy_data(self) -> Dict[str, Any]:
        """
        获取货币政策数据

        Returns:
            Dict[str, Any]: 货币政策数据
        """
        try:
            logger.info("获取货币政策数据")

            monetary_data = {}

            try:
                # LPR数据
                lpr_data = ak.macro_china_lpr()
                if not lpr_data.empty:
                    latest_lpr = lpr_data.iloc[0]
                    monetary_data["lpr"] = {
                        "value": latest_lpr.get("今值", 0),
                        "date": latest_lpr.get("日期", ""),
                        "previous": latest_lpr.get("前值", 0)
                    }
            except Exception as e:
                logger.warning(f"获取LPR数据失败: {e}")

            try:
                # 存款准备金率
                reserve_data = ak.macro_china_reserve_requirement_ratio()
                if not reserve_data.empty:
                    latest_reserve = reserve_data.iloc[0]
                    monetary_data["reserve_ratio"] = {
                        "date": latest_reserve.get("生效日期", ""),
                        "large_institutions": latest_reserve.get("大型金融机构", 0),
                        "small_institutions": latest_reserve.get("中小金融机构", 0)
                    }
            except Exception as e:
                logger.warning(f"获取存款准备金率数据失败: {e}")

            try:
                # SHIBOR数据
                shibor_data = ak.macro_china_shibor_all()
                if not shibor_data.empty:
                    latest_shibor = shibor_data.iloc[0]
                    monetary_data["shibor"] = {
                        "overnight": latest_shibor.get("Shibor隔夜", 0),
                        "1w": latest_shibor.get("Shibor1周", 0),
                        "2w": latest_shibor.get("Shibor2周", 0),
                        "1m": latest_shibor.get("Shibor1月", 0),
                        "3m": latest_shibor.get("Shibor3月", 0),
                        "date": latest_shibor.get("日期", "")
                    }
            except Exception as e:
                logger.warning(f"获取SHIBOR数据失败: {e}")

            logger.info(f"成功获取货币政策数据: {len(monetary_data)}项")
            return monetary_data

        except Exception as e:
            logger.error(f"获取货币政策数据失败: {e}")
            return {}

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_fiscal_policy_data(self) -> Dict[str, Any]:
        """
        获取财政政策数据

        Returns:
            Dict[str, Any]: 财政政策数据
        """
        try:
            logger.info("获取财政政策数据")

            fiscal_data = {}

            try:
                # 财政收入数据
                fiscal_revenue_data = ak.macro_china_czsr()
                if not fiscal_revenue_data.empty:
                    latest_revenue = fiscal_revenue_data.iloc[0]
                    fiscal_data["fiscal_revenue"] = {
                        "current_month": latest_revenue.get("当月", 0),
                        "yoy_growth": latest_revenue.get("当月同比增长", 0),
                        "cumulative": latest_revenue.get("累计", 0),
                        "cumulative_yoy": latest_revenue.get("累计同比增长", 0),
                        "date": latest_revenue.get("月份", "")
                    }
            except Exception as e:
                logger.warning(f"获取财政收入数据失败: {e}")

            try:
                # 固定资产投资数据
                investment_data = ak.macro_china_gdzctz()
                if not investment_data.empty:
                    latest_investment = investment_data.iloc[0]
                    fiscal_data["fixed_investment"] = {
                        "cumulative_yoy": latest_investment.get("累计同比增长", 0),
                        "date": latest_investment.get("月份", "")
                    }
            except Exception as e:
                logger.warning(f"获取固定资产投资数据失败: {e}")

            try:
                # 新增信贷数据
                credit_data = ak.macro_china_new_financial_credit()
                if not credit_data.empty:
                    latest_credit = credit_data.iloc[0]
                    fiscal_data["new_credit"] = {
                        "value": latest_credit.get("新增信贷", 0),
                        "date": latest_credit.get("月份", "")
                    }
            except Exception as e:
                logger.warning(f"获取新增信贷数据失败: {e}")

            logger.info(f"成功获取财政政策数据: {len(fiscal_data)}项")
            return fiscal_data

        except Exception as e:
            logger.error(f"获取财政政策数据失败: {e}")
            return {}

    def get_stock_news_for_macro_analysis(self, symbol: str, max_news: int = 100) -> List[Dict[str, Any]]:
        """
        获取股票新闻用于宏观分析

        Args:
            symbol: 股票代码
            max_news: 最大新闻数量

        Returns:
            List[Dict[str, Any]]: 新闻列表
        """
        try:
            logger.info(f"获取股票 {symbol} 新闻用于宏观分析，最多 {max_news} 条")

            # 使用现有的get_stock_news方法
            news_items = self.get_stock_news(symbol, max_news)

            news_list = []
            for news_item in news_items:
                news_dict = {
                    "title": news_item.title,
                    "content": news_item.content,
                    "publish_time": news_item.publish_time.strftime('%Y-%m-%d %H:%M:%S') if news_item.publish_time else "",
                    "source": news_item.source,
                    "url": news_item.url,
                    "keyword": ""
                }
                news_list.append(news_dict)

            logger.info(f"成功转换新闻格式: {len(news_list)}条")
            return news_list

        except Exception as e:
            logger.error(f"获取股票新闻用于宏观分析失败: {e}")
            return []

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_financial_statements(self, symbol: str) -> List[Dict[str, Any]]:
        """
        获取财务报表数据

        Args:
            symbol: 股票代码

        Returns:
            List[Dict[str, Any]]: 财务报表数据列表
        """
        try:
            logger.info(f"获取股票 {symbol} 的财务报表...")

            # 获取资产负债表数据
            logger.info("获取资产负债表...")
            try:
                # 使用自动前缀转换函数
                stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
                balance_sheet = ak.stock_financial_report_sina(
                    stock=stock_symbol_with_prefix, symbol="资产负债表")
                if not balance_sheet.empty:
                    latest_balance = balance_sheet.iloc[0]
                    previous_balance = balance_sheet.iloc[1] if len(
                        balance_sheet) > 1 else balance_sheet.iloc[0]
                    logger.info("✓ 资产负债表获取成功")
                else:
                    logger.warning("资产负债表数据为空")
                    latest_balance = pd.Series()
                    previous_balance = pd.Series()
            except Exception as e:
                logger.warning(f"获取资产负债表失败: {e}")
                latest_balance = pd.Series()
                previous_balance = pd.Series()

            # 获取利润表数据
            logger.info("获取利润表...")
            try:
                # 使用自动前缀转换函数
                stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
                income_statement = ak.stock_financial_report_sina(
                    stock=stock_symbol_with_prefix, symbol="利润表")
                if not income_statement.empty:
                    latest_income = income_statement.iloc[0]
                    previous_income = income_statement.iloc[1] if len(
                        income_statement) > 1 else income_statement.iloc[0]
                    logger.info("✓ 利润表获取成功")
                else:
                    logger.warning("利润表数据为空")
                    latest_income = pd.Series()
                    previous_income = pd.Series()
            except Exception as e:
                logger.warning(f"获取利润表失败: {e}")
                latest_income = pd.Series()
                previous_income = pd.Series()

            # 获取现金流量表数据
            logger.info("获取现金流量表...")
            try:
                # 使用自动前缀转换函数
                stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
                cash_flow = ak.stock_financial_report_sina(
                    stock=stock_symbol_with_prefix, symbol="现金流量表")
                if not cash_flow.empty:
                    latest_cash_flow = cash_flow.iloc[0]
                    previous_cash_flow = cash_flow.iloc[1] if len(
                        cash_flow) > 1 else cash_flow.iloc[0]
                    logger.info("✓ 现金流量表获取成功")
                else:
                    logger.warning("现金流量表数据为空")
                    latest_cash_flow = pd.Series()
                    previous_cash_flow = pd.Series()
            except Exception as e:
                logger.warning(f"获取现金流量表失败: {e}")
                latest_cash_flow = pd.Series()
                previous_cash_flow = pd.Series()

            # 构建财务数据
            line_items = []
            try:
                # 处理最新期间数据
                current_item = {
                    # 从利润表获取
                    "net_income": self._safe_float(latest_income.get("净利润", 0)),
                    "operating_revenue": self._safe_float(latest_income.get("营业总收入", 0)),
                    "operating_profit": self._safe_float(latest_income.get("营业利润", 0)),

                    # 从资产负债表计算营运资金
                    "working_capital": self._safe_float(latest_balance.get("流动资产合计", 0)) - self._safe_float(latest_balance.get("流动负债合计", 0)),

                    # 从现金流量表获取
                    "depreciation_and_amortization": self._safe_float(latest_cash_flow.get("固定资产折旧、油气资产折耗、生产性生物资产折旧", 0)),
                    "capital_expenditure": abs(self._safe_float(latest_cash_flow.get("购建固定资产、无形资产和其他长期资产所支付的现金", 0))),
                    "free_cash_flow": self._safe_float(latest_cash_flow.get("经营活动产生的现金流量净额", 0)) - abs(self._safe_float(latest_cash_flow.get("购建固定资产、无形资产和其他长期资产所支付的现金", 0)))
                }
                line_items.append(current_item)
                logger.info("✓ 最新期间数据处理成功")

                # 处理上一期间数据
                previous_item = {
                    "net_income": self._safe_float(previous_income.get("净利润", 0)),
                    "operating_revenue": self._safe_float(previous_income.get("营业总收入", 0)),
                    "operating_profit": self._safe_float(previous_income.get("营业利润", 0)),
                    "working_capital": self._safe_float(previous_balance.get("流动资产合计", 0)) - self._safe_float(previous_balance.get("流动负债合计", 0)),
                    "depreciation_and_amortization": self._safe_float(previous_cash_flow.get("固定资产折旧、油气资产折耗、生产性生物资产折旧", 0)),
                    "capital_expenditure": abs(self._safe_float(previous_cash_flow.get("购建固定资产、无形资产和其他长期资产所支付的现金", 0))),
                    "free_cash_flow": self._safe_float(previous_cash_flow.get("经营活动产生的现金流量净额", 0)) - abs(self._safe_float(previous_cash_flow.get("购建固定资产、无形资产和其他长期资产所支付的现金", 0)))
                }
                line_items.append(previous_item)
                logger.info("✓ 上一期间数据处理成功")

            except Exception as e:
                logger.error(f"处理财务数据时出错: {e}")
                default_item = {
                    "net_income": 0,
                    "operating_revenue": 0,
                    "operating_profit": 0,
                    "working_capital": 0,
                    "depreciation_and_amortization": 0,
                    "capital_expenditure": 0,
                    "free_cash_flow": 0
                }
                line_items = [default_item, default_item]

            logger.info(f"成功获取股票 {symbol} 财务报表数据")
            return line_items

        except Exception as e:
            logger.error(f"获取财务报表失败: {e}")
            default_item = {
                "net_income": 0,
                "operating_revenue": 0,
                "operating_profit": 0,
                "working_capital": 0,
                "depreciation_and_amortization": 0,
                "capital_expenditure": 0,
                "free_cash_flow": 0
            }
            return [default_item, default_item]

    def calculate_technical_indicators(self, prices: List[StockPrice]) -> TechnicalIndicators:
        """
        计算技术指标

        Args:
            prices: 价格数据列表

        Returns:
            TechnicalIndicators: 完整的技术指标数据
        """
        if not prices or len(prices) < 20:
            logger.warning("价格数据不足，无法计算技术指标")
            return TechnicalIndicators()

        # 转换为DataFrame
        df = pd.DataFrame([{
            'date': p.date,
            'open': p.open,
            'close': p.close,
            'high': p.high,
            'low': p.low,
            'volume': p.volume
        } for p in prices])

        # 按日期排序
        df = df.sort_values('date').reset_index(drop=True)

        try:
            # === 基础技术指标 ===
            # 计算移动平均线
            df['sma5'] = df['close'].rolling(window=5).mean()
            df['sma10'] = df['close'].rolling(window=10).mean()
            df['sma20'] = df['close'].rolling(window=20).mean()
            df['sma60'] = df['close'].rolling(window=60).mean()
            df['ema8'] = df['close'].ewm(span=8, adjust=False).mean()
            df['ema21'] = df['close'].ewm(span=21, adjust=False).mean()
            df['ema55'] = df['close'].ewm(span=55, adjust=False).mean()

            # 计算RSI
            delta = df['close'].diff()
            gain = (delta.where(delta > 0, 0)).fillna(0).rolling(window=14).mean()
            loss = (-delta.where(delta < 0, 0)).fillna(0).rolling(window=14).mean()
            rs = gain / loss
            df['rsi'] = 100 - (100 / (1 + rs))

            # 计算MACD
            exp1 = df['close'].ewm(span=12, adjust=False).mean()
            exp2 = df['close'].ewm(span=26, adjust=False).mean()
            df['macd'] = exp1 - exp2
            df['signal_line'] = df['macd'].ewm(span=9, adjust=False).mean()
            df['macd_histogram'] = df['macd'] - df['signal_line']

            # 计算布林带
            df['bb_middle'] = df['close'].rolling(window=20).mean()
            bb_std = df['close'].rolling(window=20).std()
            df['bb_upper'] = df['bb_middle'] + (bb_std * 2)
            df['bb_lower'] = df['bb_middle'] - (bb_std * 2)

            # === 新增：高级技术指标 ===

            # 1. 动量指标
            df["momentum_1w"] = df["close"].pct_change(periods=5)  # 5个交易日约等于1周
            df["momentum_1m"] = df["close"].pct_change(periods=20)  # 20个交易日约等于1个月
            df["momentum_3m"] = df["close"].pct_change(periods=60)  # 60个交易日约等于3个月
            df["momentum_6m"] = df["close"].pct_change(periods=120)  # 120个交易日约等于6个月

            # 计算成交量动量（相对于20日平均成交量的变化）
            df["volume_ma20"] = df["volume"].rolling(window=20).mean()
            df["volume_ma5"] = df["volume"].rolling(window=5).mean()
            df["volume_ratio_5"] = df["volume"] / df["volume_ma5"]
            df["volume_ratio_20"] = df["volume"] / df["volume_ma20"]

            # 2. 波动率指标
            # 历史波动率 (20日)
            returns = df["close"].pct_change()
            df["historical_volatility"] = returns.rolling(window=20).std() * np.sqrt(252)  # 年化

            # 波动率区间 (相对于过去120天的波动率的位置)
            vol_min_120d = df["historical_volatility"].rolling(window=120).min()
            vol_max_120d = df["historical_volatility"].rolling(window=120).max()
            vol_range = vol_max_120d - vol_min_120d
            df["volatility_regime"] = np.where(
                vol_range > 0,
                (df["historical_volatility"] - vol_min_120d) / vol_range,
                0.5  # 修复：如果波动范围为0，意味着波动率一直不变，应处于中间位置(50%)
            )

            vol_mean_120d = df["historical_volatility"].rolling(window=120).mean()
            vol_std_120d = df["historical_volatility"].rolling(window=120).std()
            df["volatility_z_score"] = np.where(
                vol_std_120d > 0,
                (df["historical_volatility"] - vol_mean_120d) / vol_std_120d,
                0 # 如果标准差为0，说明没有波动，Z分数为0
            )

            # ATR比率
            tr = pd.DataFrame()
            tr["h-l"] = df["high"] - df["low"]
            tr["h-pc"] = abs(df["high"] - df["close"].shift(1))
            tr["l-pc"] = abs(df["low"] - df["close"].shift(1))
            tr["tr"] = tr[["h-l", "h-pc", "l-pc"]].max(axis=1)
            df["atr"] = tr["tr"].rolling(window=14).mean()
            df["atr_ratio"] = df["atr"] / df["close"]

            # 3. 统计套利指标
            # 赫斯特指数计算
            def calculate_hurst(price_array):
                """
                标准R/S分析法计算赫斯特指数
                输入：价格数组（numpy array）
                """
                try:
                    # rolling().apply()传入的是numpy array，需要先处理
                    if len(price_array) < 50:
                        return np.nan
                    
                    # 移除NaN值
                    price_array = price_array[~np.isnan(price_array)]
                    if len(price_array) < 50:
                        return np.nan

                    # 计算对数收益率（只计算一次）
                    if len(price_array) < 2:
                        return np.nan
                    
                    log_returns = np.diff(np.log(price_array))
                    N = len(log_returns)
                    
                    if N < 30:
                        return np.nan

                    # 设置lag范围
                    max_lag = min(N//4, 50)
                    lags = range(10, max_lag + 1)
                    rs_values = []

                    for lag in lags:
                        if lag >= N:
                            continue
                        
                        rs_list = []
                        # 使用数组切片（不是pandas的iloc）
                        for i in range(N - lag + 1):
                            sub_returns = log_returns[i:i+lag] 
                            
                            if len(sub_returns) == 0:
                                continue
                            
                            # R/S分析的标准步骤
                            mean_ret = np.mean(sub_returns)
                            cumsum_dev = np.cumsum(sub_returns - mean_ret)
                            
                            R = np.max(cumsum_dev) - np.min(cumsum_dev)
                            S = np.std(sub_returns, ddof=1)
                            
                            if S > 0 and R > 0:  # 确保R和S都为正
                                rs_list.append(R / S)
                        
                        if rs_list:
                            rs_values.append(np.mean(rs_list))
                    
                    if len(rs_values) < 3:
                        return np.nan
                    
                    # 线性回归：log(R/S) vs log(lag)
                    valid_lags = lags[:len(rs_values)]
                    log_lags = np.log(valid_lags)
                    log_rs = np.log(rs_values)
                    
                    # 移除无效值
                    valid_mask = ~(np.isnan(log_rs) | np.isinf(log_rs) | (log_rs <= 0))
                    if np.sum(valid_mask) < 3:
                        return np.nan
                    
                    log_lags_clean = log_lags[valid_mask]
                    log_rs_clean = log_rs[valid_mask]
                    
                    # 线性回归得到Hurst指数
                    coeffs = np.polyfit(log_lags_clean, log_rs_clean, 1)
                    hurst = coeffs[0]
                    
                    # Hurst指数有效范围检查
                    if 0 <= hurst <= 1:
                        return hurst
                    else:
                        return np.nan
                        
                except Exception as e:
                    return np.nan

            df["hurst_exponent"] = df["close"].rolling(
                window=120,
                min_periods=60
            ).apply(calculate_hurst)

            # 偏度 (20日)
            df["skewness"] = returns.rolling(window=20).skew()

            # 峰度 (20日)
            df["kurtosis"] = returns.rolling(window=20).kurt()

            # 取最新值
            latest = df.iloc[-1]

            indicators = TechnicalIndicators(
                # 基础指标
                sma5=self._safe_float(latest.get('sma5')),
                sma10=self._safe_float(latest.get('sma10')),
                sma20=self._safe_float(latest.get('sma20')),
                sma60=self._safe_float(latest.get('sma60')),
                ema8=self._safe_float(latest.get('ema8')),
                ema21=self._safe_float(latest.get('ema21')),
                ema55=self._safe_float(latest.get('ema55')),
                rsi=self._safe_float(latest.get('rsi')),
                macd=self._safe_float(latest.get('macd')),
                signal_line=self._safe_float(latest.get('signal_line')),
                macd_histogram=self._safe_float(latest.get('macd_histogram')),
                bb_upper=self._safe_float(latest.get('bb_upper')),
                bb_middle=self._safe_float(latest.get('bb_middle')),
                bb_lower=self._safe_float(latest.get('bb_lower')),

                # 高级动量指标
                momentum_1w=self._safe_float(latest.get('momentum_1w')),
                momentum_1m=self._safe_float(latest.get('momentum_1m')),
                momentum_3m=self._safe_float(latest.get('momentum_3m')),
                momentum_6m=self._safe_float(latest.get('momentum_6m')),
                volume_ma5=self._safe_float(latest.get('volume_ma5')),
                volume_ma20=self._safe_float(latest.get('volume_ma20')),
                volume_ratio_5=self._safe_float(latest.get('volume_ratio_5')),
                volume_ratio_20=self._safe_float(latest.get('volume_ratio_20')),

                # 高级波动率指标
                historical_volatility=self._safe_float(latest.get('historical_volatility')),
                volatility_regime=self._safe_float(latest.get('volatility_regime')),
                volatility_z_score=self._safe_float(latest.get('volatility_z_score')),
                atr_ratio=self._safe_float(latest.get('atr_ratio')),

                # 统计套利指标
                hurst_exponent=self._safe_float(latest.get('hurst_exponent')),
                skewness=self._safe_float(latest.get('skewness')),
                kurtosis=self._safe_float(latest.get('kurtosis'))
            )

            logger.info("完整技术指标计算完成（包含所有高级指标）")
            return indicators

        except Exception as e:
            logger.error(f"计算技术指标失败: {e}")
            return TechnicalIndicators()
    
    def _safe_float(self, value: Any) -> Optional[float]:
        """安全转换为浮点数"""
        try:
            if pd.isna(value) or value is None:
                return None
            return float(value)
        except (ValueError, TypeError):
            return None

    def _get_stock_symbol_with_prefix(self, symbol: str) -> str:
        """
        自动获取带有交易所前缀的股票代码

        Args:
            symbol: 股票代码，如 "000001"

        Returns:
            str: 带前缀的股票代码，如 "sh000001" 或 "sz000001"
        """
        try:
            prefixed_symbol = ak.stock_a_code_to_symbol(symbol)
            logger.debug(f"股票代码 {symbol} 转换为 {prefixed_symbol}")
            return prefixed_symbol
        except Exception as e:
            logger.warning(f"股票代码转换失败 {symbol}: {e}")
            # 如果转换失败，使用简单的规则判断
            if symbol.startswith('6'):
                # 6开头的是上海交易所
                return f"sh{symbol}"
            else:
                # 其他的默认为深圳交易所
                return f"sz{symbol}"
    
    def collect_comprehensive_data(
        self, 
        ticker: str, 
        start_date: str, 
        end_date: Optional[str] = None,
        news_count: int = 10
    ) -> DataCollectionSummary:
        """
        综合数据收集
        
        Args:
            ticker: 股票代码
            start_date: 开始日期
            end_date: 结束日期
            news_count: 新闻数量
            
        Returns:
            DataCollectionSummary: 数据收集摘要
        """
        logger.info(f"开始收集股票 {ticker} 的综合数据")
        
        data_collected = {
            "price_data": False, #价格数据
            "realtime_data": False, #实时数据
            "financial_data": False, #财务数据
            "news_data": False, #新闻数据
            "technical_indicators": False #技术指标
        }
        
        price_records_count = 0 #价格数据条数
        news_count_actual = 0 #新闻数据条数
        
        try:
            # 收集价格数据
            price_data = self.get_stock_hist_data(ticker, start_date, end_date)
            if price_data:
                data_collected["price_data"] = True
                price_records_count = len(price_data)
          
            # 收集实时数据
            realtime_data = self.get_stock_realtime_data(ticker)
            if realtime_data:
                data_collected["realtime_data"] = True
            
            # 收集财务数据（现在包含财务报表）
            financial_data = self.get_financial_indicators(ticker)
            if financial_data:
                data_collected["financial_data"] = True
            
            # 收集新闻数据
            news_data = self.get_stock_news(ticker, news_count)
            if news_data:
                data_collected["news_data"] = True
                news_count_actual = len(news_data)
            
            # 计算技术指标
            if price_data:
                technical_indicators = self.calculate_technical_indicators(price_data)
                if technical_indicators:
                    data_collected["technical_indicators"] = True
            
            # 生成摘要
            collected_count = sum(data_collected.values())
            total_count = len(data_collected)
            
            summary = f"数据收集完成: {collected_count}/{total_count} 项成功"
            if price_records_count > 0:
                summary += f", 价格数据 {price_records_count} 条"
            if news_count_actual > 0:
                summary += f", 新闻数据 {news_count_actual} 条"
            
            return DataCollectionSummary(
                ticker=ticker,
                start_date=start_date,
                end_date=end_date or datetime.now().strftime("%Y%m%d"),
                data_collected=data_collected,
                price_records_count=price_records_count,
                news_count=news_count_actual,
                summary=summary,
                price_data=price_data,# 增补返回信息
                realtime_data=realtime_data,
                financial_data=financial_data,
                news_data=news_data,
                technical_indicators=technical_indicators
            )
            
        except Exception as e:
            logger.error(f"综合数据收集失败: {e}")
            return DataCollectionSummary(
                ticker=ticker,
                start_date=start_date,
                end_date=end_date or datetime.now().strftime("%Y%m%d"),
                data_collected=data_collected,
                price_records_count=price_records_count,
                news_count=news_count_actual,
                summary=f"数据收集失败: {str(e)}"
            )

    # ==================== 缓存管理方法 ====================

    def clear_cache(self, pattern: str = None):
        """
        清除缓存

        Args:
            pattern: 缓存键模式，如果为None则清除所有缓存
        """
        self.cache_manager.invalidate(pattern)
        logger.info(f"缓存已清除: {pattern or '全部'}")

    def get_cache_stats(self) -> Dict[str, Any]:
        """获取缓存统计信息"""
        return self.cache_manager.get_stats()

    def print_cache_stats(self):
        """打印缓存统计信息"""
        self.cache_manager.print_stats()

    def clear_realtime_cache(self):
        """清除实时数据缓存"""
        self.clear_cache("get_stock_realtime_data")
        self.clear_cache("_get_cached_realtime_data")
        logger.info("实时数据缓存已清除")

    def clear_financial_cache(self, symbol: str = None):
        """清除财务数据缓存"""
        if symbol:
            self.clear_cache(f"get_financial_indicators:{symbol}")
            self.clear_cache(f"get_financial_statements:{symbol}")
        else:
            self.clear_cache("get_financial_indicators")
            self.clear_cache("get_financial_statements")
        logger.info(f"财务数据缓存已清除: {symbol or '全部'}")

    def clear_news_cache(self, symbol: str = None):
        """清除新闻数据缓存"""
        if symbol:
            self.clear_cache(f"get_stock_news:{symbol}")
        else:
            self.clear_cache("get_stock_news")
        logger.info(f"新闻数据缓存已清除: {symbol or '全部'}")

    # ==================== 估值分析专用方法 ====================

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_balance_sheet(self, symbol: str) -> Optional[Dict[str, Any]]:
        """
        获取资产负债表数据

        Args:
            symbol: 股票代码

        Returns:
            Optional[Dict[str, Any]]: 资产负债表数据
        """
        try:
            logger.info(f"获取股票 {symbol} 的资产负债表...")

            # 东方财富资产负债表
            stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
            balance_sheet = ak.stock_balance_sheet_by_report_em(symbol=stock_symbol_with_prefix)

            if balance_sheet.empty:
                logger.warning(f"未获取到股票 {symbol} 的资产负债表数据")
                return None

            # 转换为字典格式，取最新的数据
            latest_data = balance_sheet.iloc[0].to_dict()

            logger.info(f"✓ 成功获取股票 {symbol} 资产负债表数据")
            return latest_data

        except Exception as e:
            logger.error(f"获取资产负债表失败: {e}")
            return None

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_income_statement(self, symbol: str) -> Optional[Dict[str, Any]]:
        """
        获取利润表数据

        Args:
            symbol: 股票代码

        Returns:
            Optional[Dict[str, Any]]: 利润表数据
        """
        try:
            logger.info(f"获取股票 {symbol} 的利润表...")

            # 东方财富利润表
            stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
            income_statement = ak.stock_profit_sheet_by_report_em(symbol=stock_symbol_with_prefix)

            if income_statement.empty:
                logger.warning(f"未获取到股票 {symbol} 的利润表数据")
                return None

            # 转换为字典格式，取最新的数据
            latest_data = income_statement.iloc[0].to_dict()

            logger.info(f"✓ 成功获取股票 {symbol} 利润表数据")
            return latest_data

        except Exception as e:
            logger.error(f"获取利润表失败: {e}")
            return None

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_cash_flow_statement(self, symbol: str) -> Optional[Dict[str, Any]]:
        """
        获取现金流量表数据

        Args:
            symbol: 股票代码

        Returns:
            Optional[Dict[str, Any]]: 现金流量表数据
        """
        try:
            logger.info(f"获取股票 {symbol} 的现金流量表...")

            # 东方财富现金流量表
            stock_symbol_with_prefix = self._get_stock_symbol_with_prefix(symbol)
            cash_flow = ak.stock_cash_flow_sheet_by_report_em(symbol=stock_symbol_with_prefix)

            if cash_flow.empty:
                logger.warning(f"未获取到股票 {symbol} 的现金流量表数据")
                return None

            # 转换为字典格式，取最新的数据
            latest_data = cash_flow.iloc[0].to_dict()

            logger.info(f"✓ 成功获取股票 {symbol} 现金流量表数据")
            return latest_data

        except Exception as e:
            logger.error(f"获取现金流量表失败: {e}")
            return None

    def _get_current_time(self) -> str:
        """获取当前时间字符串"""
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_stock_industry_info(self, symbol: str) -> Optional[Dict[str, Any]]:
        """
        获取股票行业信息

        Args:
            symbol: 股票代码

        Returns:
            Optional[Dict[str, Any]]: 股票行业信息
        """
        try:
            logger.info(f"获取股票 {symbol} 的行业信息...")

            # 个股信息查询
            stock_info = ak.stock_individual_info_em(symbol=symbol)

            if stock_info.empty:
                logger.warning(f"未获取到股票 {symbol} 的基本信息")
                return None

            # 提取行业相关信息
            industry_info = {}
            for _, row in stock_info.iterrows():
                item = row['item']
                value = row['value']

                if '行业' in item:
                    industry_info[item] = value
                elif '板块' in item:
                    industry_info[item] = value
                elif '概念' in item:
                    industry_info[item] = value

            logger.info(f"✓ 成功获取股票 {symbol} 行业信息")
            return industry_info

        except Exception as e:
            logger.error(f"获取股票行业信息失败: {e}")
            return None

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_industry_stocks(self, industry_name: str) -> Optional[List[str]]:
        """
        获取指定行业的成份股

        Args:
            industry_name: 行业名称

        Returns:
            Optional[List[str]]: 行业成份股代码列表
        """
        try:
            logger.info(f"获取行业 {industry_name} 的成份股...")

            # 行业板块成份股
            industry_stocks = ak.stock_board_industry_cons_em(symbol=industry_name)

            if industry_stocks.empty:
                logger.warning(f"未获取到行业 {industry_name} 的成份股")
                return None

            # 提取股票代码列表
            stock_codes = industry_stocks['代码'].tolist()

            logger.info(f"✓ 成功获取行业 {industry_name} 成份股，共 {len(stock_codes)} 只")
            return stock_codes

        except Exception as e:
            logger.error(f"获取行业成份股失败: {e}")
            return None

    @cached(cache_level=CacheLevel.LONG)  # 2小时缓存
    @backoff.on_exception(backoff.expo, Exception, max_tries=3)
    def get_all_industry_names(self) -> Optional[List[str]]:
        """
        获取所有行业板块名称

        Returns:
            Optional[List[str]]: 行业名称列表
        """
        try:
            logger.info("获取所有行业板块名称...")

            # 行业板块名称
            industry_names = ak.stock_board_industry_name_em()

            if industry_names.empty:
                logger.warning("未获取到行业板块名称")
                return None

            # 提取行业名称列表
            names = industry_names['板块名称'].tolist()

            logger.info(f"✓ 成功获取所有行业板块名称，共 {len(names)} 个")
            return names

        except Exception as e:
            logger.error(f"获取行业板块名称失败: {e}")
            return None

    def get_industry_peers(self, symbol: str, max_peers: int = 10) -> List[str]:
        """
        获取同行业可比公司

        Args:
            symbol: 目标股票代码
            max_peers: 最大返回同行业公司数量

        Returns:
            List[str]: 同行业可比公司代码列表
        """
        try:
            logger.info(f"开始获取股票 {symbol} 的同行业可比公司...")

            # 1. 获取目标股票的行业信息
            industry_info = self.get_stock_industry_info(symbol)
            if not industry_info:
                logger.warning(f"无法获取股票 {symbol} 的行业信息，使用降级策略")
                return self._get_fallback_peers(symbol, max_peers)

            # 2. 从行业信息中提取主要行业名称
            target_industry = None
            for key, value in industry_info.items():
                if '行业' in key and value:
                    target_industry = value
                    break

            if not target_industry:
                logger.warning(f"无法确定股票 {symbol} 的主要行业，使用降级策略")
                return self._get_fallback_peers(symbol, max_peers)

            logger.info(f"股票 {symbol} 属于行业: {target_industry}")

            # 3. 获取该行业的所有成份股
            industry_stocks = self.get_industry_stocks(target_industry)
            if not industry_stocks:
                logger.warning(f"无法获取行业 {target_industry} 的成份股，使用降级策略")
                return self._get_fallback_peers(symbol, max_peers)

            # 4. 移除目标股票本身
            if symbol in industry_stocks:
                industry_stocks.remove(symbol)

            # 5. 筛选和排序可比公司
            peers = self._filter_and_rank_peers(symbol, industry_stocks, max_peers)

            logger.info(f"✓ 成功获取股票 {symbol} 的同行业可比公司: {peers}")
            return peers

        except Exception as e:
            logger.error(f"获取同行业可比公司失败: {e}")
            return self._get_fallback_peers(symbol, max_peers)

    def _filter_and_rank_peers(self, target_symbol: str, industry_stocks: List[str], max_peers: int) -> List[str]:
        """
        筛选和排序可比公司

        Args:
            target_symbol: 目标股票代码
            industry_stocks: 行业内所有股票代码
            max_peers: 最大返回数量

        Returns:
            List[str]: 筛选后的可比公司列表
        """
        try:
            # 简化的筛选逻辑：
            # 1. 优先选择主板股票（6开头和0开头）
            # 2. 避免ST股票
            # 3. 限制数量

            filtered_peers = []

            # 优先级排序：主板 > 中小板 > 创业板
            priority_order = []

            for stock in industry_stocks:
                # 跳过ST股票（简化判断）
                if 'ST' in stock:
                    continue

                # 按板块分类
                if stock.startswith('60'):  # 上海主板
                    priority_order.insert(0, stock)
                elif stock.startswith('00'):  # 深圳主板
                    priority_order.insert(len([s for s in priority_order if s.startswith('60')]), stock)
                elif stock.startswith('30'):  # 创业板
                    priority_order.append(stock)
                else:
                    priority_order.append(stock)

            # 取前max_peers个
            filtered_peers = priority_order[:max_peers]

            return filtered_peers

        except Exception as e:
            logger.error(f"筛选可比公司失败: {e}")
            return industry_stocks[:max_peers]
